/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.nhncorp.neptune.common;

import java.io.IOException;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import com.nhncorp.neptune.common.conf.NConfiguration;
import com.nhncorp.pleiades.client.LockEventData;
import com.nhncorp.pleiades.client.LockEventHandlerFactory;
import com.nhncorp.pleiades.client.LockService;
import com.nhncorp.pleiades.client.LockServiceClient;
import com.nhncorp.pleiades.client.PleiadesUser;
import com.nhncorp.pleiades.conf.PleiadesConfiguration;
import com.nhncorp.pleiades.protocol.LockMetaData;

public class LockServiceFactory {
  public static final Log LOG = LogFactory.getLog(LockServiceFactory.class.getName());
  
  //lock owner -> lock service
  private static Map<String, LockService> lockServices = new HashMap<String, LockService>();
  private static Set<String> removedLockService = new HashSet<String>();
  
  private static PleiadesConfiguration pConf;
  
  public static LockService getLockService(NConfiguration conf, 
      String owner, boolean allowCallback) throws IOException {
    return getLockService(conf, owner, allowCallback, 0);
  }
  
  public static LockService getLockService(NConfiguration conf, 
      String owner, boolean allowCallback, int beartbeatPort) throws IOException {
      synchronized(lockServices) {
      if(!lockServices.containsKey(owner)) {
        if(removedLockService.contains(owner)) {
          return null;
        }
        PleiadesUser user = new PleiadesUser(conf.get("pleiades.service.name", "neptune"), owner, beartbeatPort);
        user.setSessionDisable(!allowCallback);
        if(pConf == null) {
          pConf = new PleiadesConfiguration();

          copyProperty(conf, pConf, "pleiades.cell.members");
          copyProperty(conf, pConf, "clusterManager.port");
          copyProperty(conf, pConf, "clusterID");
        }
        LockServiceClient lockService = new LockServiceClient(user, pConf, beartbeatPort != 0);
        if(allowCallback) {
          lockService.setEventHandler(LockEventHandlerFactory.newInstance(pConf));
        }
        lockServices.put(owner, lockService);
        
        return lockService;
      } else {
        return lockServices.get(owner);
      }
    }
  }

  private static void copyProperty(NConfiguration conf, PleiadesConfiguration pConf, String key) {
    String nConfValue = conf.get(key);
    if(nConfValue != null && nConfValue.length() > 0) {
      pConf.set(key, nConfValue);
    }    
  }
  
  public static void close(String owner) {
    close(owner, false);
  }
  
  public static void close(String owner, boolean canGetAgain) {
    synchronized(lockServices) {
      if(lockServices.containsKey(owner)) {
        //LOG.debug("=====LockService closed[" + owner + "]=======");
        LockService lockService = lockServices.get(owner);
        try {
          lockService.releaseAll();
        } catch (IOException e) {
          LOG.error("fail to release all", e);
        }
        lockService.close();
//        lockService.poison(true);
        lockServices.remove(owner);
        if(!canGetAgain) {
          removedLockService.add(owner);
        }
      }
    }
  }
  
  public static void clear(String owner) {
    synchronized(lockServices) {
      LockService lockService = lockServices.remove(owner);
      if(lockService != null) {
        try {
          lockService.releaseAll();
        } catch (IOException e) {
          LOG.error("fail to release all", e);
        }
        lockService.close();
//        lockService.poison(true);
      }
      removedLockService.remove(owner);
    }
  }
  
  public static void clear() {
    LOG.debug("=====LockService clear=======");
    synchronized(lockServices) { 
      for(LockService eachLockService: lockServices.values()) {
//        eachLockService.releaseAll();
        eachLockService.close();
//        eachLockService.poison(true);
      }
      lockServices.clear();
      removedLockService.clear();
    }
  }
  
  public static String getLockMetaDataPathName(LockMetaData lockMetaData) {
    String lockPath = lockMetaData.getLockId();
    int index = lockPath.lastIndexOf("/");
    if(index < 0) {
      return null;
    }
    
    return lockPath.substring(index + 1);
  }
  
  public static String getLockPathName(LockEventData eventData) {
    String lockPath = eventData.getLockId();
    int index = lockPath.lastIndexOf("/");
    if(index < 0) {
      return null;
    }
    
    return lockPath.substring(index + 1);
  }
}
